package org.leiyang.common.holder;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import org.leiyang.common.dtos.ConnectedMsgServerInfo;
import org.leiyang.common.dtos.MsgServerInfo;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 管理客户端和服务端的连接关系
 * 一个客户端在同一时刻只会与一个服务端建立长连接
 */
@Component
public class MsgServerAndClientHolder {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private final ReactiveRedisCache redisCache;
    private OnlineServerHolder onlineServerHolder;

    /**
     * 缓存在线客户端与server的路由信息
     * 格式: client_server_key + {ClientId}
     */
    private final static String client_server_key = "online:c:s:";

    /**
     * 缓存客户端和当前Server的连接关系
     * ClientId => MsgServerInfo
     * ClientId = {userId}@{clientIp}
     */
    private static final Map<String, ConnectedMsgServerInfo> MSG_SERVER_CLIENT_HOLDER = new ConcurrentHashMap<>();

    public MsgServerAndClientHolder(ReactiveRedisCache redisCache) {
        this.redisCache = redisCache;
    }

    public OnlineServerHolder getOnlineServerHolder() {
        return onlineServerHolder;
    }
    @Autowired
    public void setOnlineServerHolder(OnlineServerHolder onlineServerHolder) {
        this.onlineServerHolder = onlineServerHolder;
    }

    /**
     * 获取指定客户端连接的服务端信息
     * 使用场景：获取指定客户端的服务器信息 或者 首次连接到server时调用
     * @param clientId 客户端ID(唯一)
     * @param serverAddress server地址IP
     * @param port 端口
     * @return 已成功连接的服务端信息, 如果ConnectedMsgServerInfo#channel为空则表示是待连接的客户端
     */
    public Mono<ConnectedMsgServerInfo> gotConnectedServerInfo(String clientId, ChannelHandlerContext ctx, String serverAddress, String port) {
        logger.info("开始执行org.leiyang.common.holder.MsgServerAndClientHolder.gotConnectedServerInfo");
        return Mono.just(MSG_SERVER_CLIENT_HOLDER.containsKey(clientId)).flatMap(contains -> {
            if(contains) {
                return Mono.just(MSG_SERVER_CLIENT_HOLDER.get(clientId));
            }
            String redisKey = client_server_key.concat(clientId);
            return redisCache.keyExist(redisKey).flatMap(exist -> {
                //Redis里有缓存，这种情况基本不会出现
                logger.info("开始查询redis里是否有ConnectedMsgServerInfo信息");
                if(exist) {
                    return redisCache.getCacheObject(redisKey).map(obj -> {
                        ConnectedMsgServerInfo connectedMsgServerInfo = (ConnectedMsgServerInfo) obj;
                        MSG_SERVER_CLIENT_HOLDER.putIfAbsent(clientId, connectedMsgServerInfo);
                        return connectedMsgServerInfo;
                    });
                }
                //表示是发起第一次连接的客户端
                logger.info("开始执行第一次连接的客户端的操作");
                Mono<MsgServerInfo> msgServerInfoMono = onlineServerHolder.joinOnlineServer(serverAddress, port);
                Mono<Boolean> joinChannelMono = onlineServerHolder.joinChannel(serverAddress, port, ctx);
                Mono<ConnectedMsgServerInfo> saveConnectedMsgServerInfoMono =msgServerInfoMono.flatMap(msInfo -> {
                    ConnectedMsgServerInfo newMsgServerInfo = ConnectedMsgServerInfo.builder()
                            .msgServerInfo(msInfo)
                            .channelId(ctx.channel().id().asLongText()) // redis里仅存储channelId
                            .build();
                    return redisCache.setCacheObject(redisKey, newMsgServerInfo).doOnNext(save -> {
                        if(save) {
                            newMsgServerInfo.setChannel(ctx.channel());
                            MSG_SERVER_CLIENT_HOLDER.putIfAbsent(clientId, newMsgServerInfo);
                        }
                    }).map(save -> newMsgServerInfo);
                });
                return joinChannelMono.flatMap(joinChannel -> saveConnectedMsgServerInfoMono);
            });
        });
    }

    /**
     * 移除客户端与server的连接
     * 使用场景: 当客户端下线时
     * @param clientId 客户端ID(唯一)
     */
    public Mono<Boolean> removeOfflineClient(String clientId) {
        final String key = client_server_key.concat(clientId);
        Mono<Long> delNumMono = redisCache.deleteObject(key);
        Mono<ConnectedMsgServerInfo> delConnectedMsgServerInfoMono = redisCache.keyExist(key).flatMap(exist -> {
            if(exist) {
                return redisCache.getCacheObject(key).map(obj -> (ConnectedMsgServerInfo) obj);
            }
            return Mono.empty();
        });
        Mono<Boolean> removeChannelMono = delConnectedMsgServerInfoMono.flatMap(removeOfflineClient -> {
            if(Objects.nonNull(removeOfflineClient)) {
                logger.info("delConnectedMsgServerInfoMono时获取到的ConnectedMsgServerInfo: {}", removeOfflineClient);
                //只能从本地缓存里获取，redis里的channel信息是空的，因为channel不能被序列化
                ConnectedMsgServerInfo localCachedServerInfo = MSG_SERVER_CLIENT_HOLDER.get(clientId);
                if(Objects.isNull(localCachedServerInfo)) {
                    logger.error("本地缓存里未找到{}的ConnectedMsgServerInfo信息，请无法移除相关ConnectedMsgServerInfo信息!", clientId);
                    return Mono.just(false);
                }
                Channel channel = localCachedServerInfo.getChannel();
                return onlineServerHolder.removeChannel(removeOfflineClient.getMsgServerInfo().getServerAddress(),
                        removeOfflineClient.getMsgServerInfo().getServerPort(),
                        channel).map(removeChannelResult -> {
                            if(removeChannelResult) {
                                if(channel.isActive()) {
                                    channel.close();
                                }
                            }
                            return removeChannelResult;
                        });
            }
            return Mono.just(true);
        });
        return removeChannelMono.flatMap(removeChannel -> {
            if(removeChannel) {
                return delNumMono.map(delNum -> {
                    if(delNum > 0) {
                        //最后再移除本地缓存里的信息
                        MSG_SERVER_CLIENT_HOLDER.remove(clientId);
                        return true;
                    }
                    return false;
                });
            }
            return Mono.just(false);
        });
    }

    /**
     * 移除指定server的所有连接
     * @param removedServerInfo server信息
     */
    public void batchRemoveByServer(MsgServerInfo removedServerInfo) {
        Set<String> deleteKeys = MSG_SERVER_CLIENT_HOLDER.entrySet().stream().filter(entry -> Objects.equals(entry.getValue().getMsgServerInfo(), removedServerInfo))
                .map(Map.Entry::getKey).collect(Collectors.toSet());
        if(deleteKeys.isEmpty()) {
            return;
        }
        redisCache.deleteKeys(deleteKeys.stream().peek(key -> {
            ConnectedMsgServerInfo serverInfo = MSG_SERVER_CLIENT_HOLDER.remove(key);
            Channel channel = serverInfo.getChannel();
            if (channel.isActive()) {
                channel.close();
            }
        }).map(client_server_key::concat).distinct().toArray(String[]::new)).subscribe();
    }

    /**
     * 确认指定的客户端是否在线
     * @param clientIp 客户端出口公网IP
     * @param uid 用户ID
     * @return 是否在线
     */
    public Mono<Boolean> confirmClientOnline(String clientIp, Long uid) {
        String clientId = uid + "@" + clientIp.replaceAll("\\.", "");
        final String key = client_server_key.concat(clientId);
        return redisCache.keyExist(key);
    }

    /**
     * 查询指定的用户连接的msgServer列表 (同一个账号最多同时开启10个客户端)
     * @param uid 用户ID
     * @return 指定的用户连接的msgServer列表
     */
    public Flux<ConnectedMsgServerInfo> confirmUserOnline(Long uid) {
        String clientIdPrefix = uid + "@";
        final String keyPrefix = client_server_key.concat(clientIdPrefix);
        return redisCache.scan(keyPrefix.concat("*"), 10L)
                .flatMap(redisCache::getCacheObject)
                .map(obj -> (ConnectedMsgServerInfo) obj);
    }

    /**
     * 获取指定用户下指定客户端IP连接的msgServer
     * @param uid 用户ID
     * @param clientIp 客户端出口公网IP
     * @return msgServer
     */
    public Mono<ConnectedMsgServerInfo> msgServerByClientIp(Long uid, String clientIp) {
        String clientId = uid + "@" + clientIp.replaceAll("\\.", "");
        logger.info("本次获取clientId:[{}]连接的msgServer.", clientId);
        return redisCache.getCacheObject(client_server_key.concat(clientId))
                .map(obj -> (ConnectedMsgServerInfo) obj);
    }

    /**
     * 根据{@link Channel}获取绑定的对应的ClientId
     */
    public String gotClientIdByChannel(Channel channel) {
        logger.info("MSG_SERVER_CLIENT_HOLDER: {}", MSG_SERVER_CLIENT_HOLDER);
        Map.Entry<String, ConnectedMsgServerInfo> selectEntry = MSG_SERVER_CLIENT_HOLDER.entrySet().stream()
                .filter(entry -> checkChannelsSame(channel, entry.getValue().getChannel()))
                .findAny()
                .orElse(null);
        return Objects.isNull(selectEntry) ? null : selectEntry.getKey();
    }

    /**
     * 确认两个{@link Channel}是否相同
     * @param c1 Channel
     * @param c2 Channel
     * @return 是否是同一个Channel
     */
    private Boolean checkChannelsSame(Channel c1, Channel c2) {
        return Objects.equals(c1.id(), c2.id()) &&
                Objects.equals(c1.remoteAddress(), c2.remoteAddress());
    }
}